package org.tao.lightningmq.broker.event.spi.listenr;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.slave.SlaveSyncService;
import org.tao.lightningmq.common.dto.BrokerMasterChangeReqDTO;
import org.tao.lightningmq.common.remote.BrokerMasterChangeCallBack;

public class BrokerMasterChangeListener implements BrokerMasterChangeCallBack {

    private static final Logger LOGGER = LoggerFactory.getLogger(BrokerMasterChangeListener.class);

    @Override
    public void onReceive(BrokerMasterChangeReqDTO reqDTO) {
        String newMasterNodeAddr = reqDTO.getNewMasterNodeAddr();

        //由于存在多个Nameserver节点，因此需要判断主从切换是否已经完成以保证幂等性
        SlaveSyncService slaveSyncService = CommonCache.getSlaveSyncService();
        if (slaveSyncService.getCurrentMasterAddr().equals(newMasterNodeAddr)) {
            LOGGER.info("主从切换已经完成，返回");
            return;
        }

        LOGGER.info("newMasterNodeAddr: {}", newMasterNodeAddr);

        // 关闭和原始的主节点的连接
        slaveSyncService.closeConnection();

        // 判断新主节点是不是自己，是的话就不用连接了
        String brokerIp = CommonCache.getGlobalProperties().getBrokerIp();
        Integer brokerPort = CommonCache.getGlobalProperties().getBrokerPort();
        if (newMasterNodeAddr.equals(brokerIp + ":" + brokerPort)) {
            LOGGER.info("当前节点成为新的Broker主节点");
            return;
        }

        // 和新节点建立连接
        slaveSyncService = new SlaveSyncService();
        CommonCache.setSlaveSyncService(slaveSyncService);
        boolean connectionStat = slaveSyncService.connectMasterBrokerNode(newMasterNodeAddr);
        if(connectionStat) {
            slaveSyncService.sendStartSyncMsg();
            LOGGER.info("和新主节点: {} 建立连接!", newMasterNodeAddr);
        }

    }
}
